package com.angel.bo.common.handler.stomp;

import com.angel.bo.common.util.thread.AngelThreadExecutorFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.messaging.simp.stomp.*;

import java.lang.reflect.Type;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;

/**
 *
 */
@Slf4j
public class StompSessionHandler extends StompSessionHandlerAdapter {

    private String topicUrl = "/user/queue/receive";

    /**
     * 接收到消息的处理动作
     */
    private StompReceiveHandler stompReceiveHandler;

    private CountDownLatch latch;

    public StompSessionHandler(StompReceiveHandler stompReceiveHandler, CountDownLatch latch) {
        this.stompReceiveHandler = stompReceiveHandler;
        this.latch = latch;
    }

    public StompSessionHandler(String topicUrl, StompReceiveHandler stompReceiveHandler, CountDownLatch latch) {
        this.topicUrl = topicUrl;
        this.stompReceiveHandler = stompReceiveHandler;
        this.latch = latch;
    }

    private StompSessionHandler(){}

    @Override
    public void handleFrame(StompHeaders headers, Object payload) {
        super.handleFrame(headers, payload);
    }

    @Override
    public void afterConnected(StompSession session, StompHeaders connectedHeaders) {
        session.subscribe(topicUrl, new StompFrameHandler() {
            @Override
            public Type getPayloadType(StompHeaders headers) {
                return String.class;
            }

            @Override
            @SuppressWarnings("ALL")
            public void handleFrame(StompHeaders headers, Object payload) {
                ThreadPoolExecutor threadPoolExecutor = AngelThreadExecutorFactory.getEcgThreadPoolExecutor();
                threadPoolExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        stompReceiveHandler.apply(payload);
                    }
                });

                log.info("接收到的消息{}",payload);
            }
        });

    }

    @Override
    public void handleException(StompSession session, StompCommand command, StompHeaders headers, byte[] payload, Throwable exception) {
        super.handleException(session, command, headers, payload, exception);
        log.error("StompError,handleException 异常如下：",exception);
        latch.countDown();
    }


    @Override
    public void handleTransportError(StompSession session, Throwable exception) {
        super.handleTransportError(session, exception);
        log.error("StompError handleTransportError,异常如下：",exception);
        latch.countDown();
    }


    public void setLatch(CountDownLatch latch) {
        this.latch = latch;
    }
}
